package org.xyf.IM.config;

import com.alibaba.cloud.nacos.NacosDiscoveryProperties;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.*;
import org.redisson.api.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.cloud.client.discovery.event.InstanceRegisteredEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.KafkaAdmin;
import org.xyf.IM.common.CacheData;
import org.xyf.IM.common.GlobalConstant;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;
import java.util.concurrent.ExecutionException;

@Configuration(proxyBeanMethods = false)
@EnableKafka
@Slf4j
public class KafkaTopicConfig implements ApplicationListener<InstanceRegisteredEvent> {

    @Value("${spring.application.name}")
    private String appName;

    @Resource
    private DiscoveryClient discoveryClient;

    @Resource
    private RedissonClient redissonClient;

    @Resource
    private KafkaAdmin kafkaAdmin;


    /**
     * 在实例注册完成后触发
     *
     * @param event
     */
    @Override
    public void onApplicationEvent(InstanceRegisteredEvent event) {
        Map<String, Object> kafkaConfig = kafkaAdmin.getConfigurationProperties();
        NacosDiscoveryProperties properties = (NacosDiscoveryProperties) event.getConfig();
        RLock lock = redissonClient.getLock(GlobalConstant.SERVICE_REGISTRY_LOCK);
        try {
            lock.lock();
            this.createAndBlindTopic(properties.getIp(), properties.getPort());
        } catch (Exception e) {
            log.error("createAndBlindTopic error : ", e);
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }
    }

    /**
     * 在实例启动时创建topic;如果topic已经存在则不会继续创建
     * topic与服务实例的对应关系维护在redis中
     */
    public void createAndBlindTopic(String ip, int port) throws ExecutionException, InterruptedException, UnknownHostException {
        RMap<String, String> rMap = redissonClient.getMap(GlobalConstant.IM_TOPIC_MAP_KEY);//不存在则会自动创建
        String mKey = ip + "::" + port;
        String topicName = rMap.get(mKey);
        if (Objects.isNull(topicName)) { //不存在说明是第一次启动服务;需要新建或者绑定原有的topic
            List<ServiceInstance> instances = discoveryClient.getInstances(appName);

            Map<String, String> notBlindTopics = this.getNoBlindTopic(instances, rMap.readAllMap());
            if (notBlindTopics.isEmpty()) {
                topicName = appName + instances.size();
                log.info("正在创建当前实例对应topic:{}", topicName);
                if (!this.isTopicExist(topicName)) {
                    NewTopic newTopic = TopicBuilder.name(topicName)
                            .partitions(1)//分区和副本都应该改为可配置的
                            .replicas(1)
                            .build();
                    kafkaAdmin.createOrModifyTopics(newTopic);
                }

            } else { //绑定那些被销毁pod所建的topic
                Map.Entry<String, String> first = notBlindTopics.entrySet().stream().findFirst().get();
                rMap.remove(first.getKey());
                topicName = first.getValue(); //绑定第一个即可
            }

            rMap.put(mKey, topicName);
        }
        CacheData.cacheTopicName(topicName);
    }

    private boolean isTopicExist(String topicName) throws ExecutionException, InterruptedException {
        AdminClient client = AdminClient.create(kafkaAdmin.getConfigurationProperties());
        ListTopicsResult topics = client.listTopics();
        Set<String> names = topics.names().get();
        log.info("topic set is : {}",names);
        client.close();;
        return names.contains(topicName);
    }

    /**
     * 存在一些情况容器被销毁了,topic还在的情况，此时应该让新启动的容器绑定这些topic
     * 判断逻辑: 遍历redis map中的key,如果某个key与instance生成的key集合所包含，则证明，此pod已经被销毁,它所对应的topic需要重新建立对应关系
     */
    private Map<String, String> getNoBlindTopic(List<ServiceInstance> instances, Map<String, String> topicMap) throws UnknownHostException {
        Map<String, String> result = new HashMap<>();
        Set<String> instanceKeys = new HashSet<>();
        for (ServiceInstance instance : instances) {
            String ip = this.getIpFromHost(instance.getHost());
            String key = ip + "::" + instance.getPort();
            instanceKeys.add(key);
        }
        topicMap.forEach((k, v) -> {
            if (!instanceKeys.contains(k)) {
                result.put(k, v);
            }
        });
        return result;
    }

    private String getIpFromHost(String host) throws UnknownHostException {
        InetAddress address = InetAddress.getByName(host);
        return address.getHostAddress();
    }
}
